package kuoge.practice.kafka.stream.member.clickrate;

import io.netty.util.HashedWheelTimer;
import io.netty.util.TimerTask;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import kuoge.practice.kafka.stream.member.KafkaConfigFactory;

import java.time.Duration;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import static org.apache.kafka.streams.KafkaStreams.State.NOT_RUNNING;
import static kuoge.practice.kafka.stream.member.TopicConstants.ITEM_CLICK_REPORT_TOPIC;
import static kuoge.practice.kafka.stream.member.TopicConstants.ITEM_CLICK_TOPIC;

/**
 * @Description
 */
public class UserClickEventConsumer {
    private static final KafkaConsumer<String, Long> kafkaConsumer;
    private final HashedWheelTimer hashedWheelTimer = new HashedWheelTimer();
    private volatile KafkaStreams kafkaStreams;
    private final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    static {
        kafkaConsumer = new KafkaConsumer<>(KafkaConfigFactory.getConsumerProperties());
    }

    public void startClickStreamProcess() {
        final StreamsBuilder streamsBuilder = new StreamsBuilder();
        // key是商品ID，value是点击次数
        final KStream<String, String> itemClickKStream = streamsBuilder.stream(ITEM_CLICK_TOPIC, Consumed.with(Serdes.String(), Serdes.String()));
        // 定义多个窗口统计周期
        final List<TimeWindows> timeWindowsList = List.of(
                TimeWindows.of(Duration.ofSeconds(10)),
                TimeWindows.of(Duration.ofSeconds(30)),
                TimeWindows.of(Duration.ofSeconds(45)),
                TimeWindows.of(Duration.ofSeconds(60))
        );
        final List<KTable<Windowed<String>, Long>> kTables = new ArrayList<>();
        timeWindowsList.forEach(timeWindows -> kTables.add(itemClickKStream
                .peek((key, value) -> System.out.printf("in the first peek transformation,  second = %s, key:%s, value:%s", LocalDateTime.now().format(dateTimeFormatter), key, value))
                .map((key, value) -> new KeyValue<>(timeWindows.sizeMs + "#" + key, "1"))
                .groupByKey(Grouped.with(Serdes.String(), Serdes.String()))
                .windowedBy(timeWindows)
                .count(Materialized.as("click_counts_" + timeWindows.size())))
        );

        kTables.forEach(windowedLongKTable -> windowedLongKTable.toStream()
                .peek((key, value) -> System.out.println(String.format("in the second peek transformation, second = %s, key:%s, value:%s", LocalDateTime.now().format(dateTimeFormatter), key.key(), value)))
                .map((k, v) -> new KeyValue<>(k.key(), v))
                .to(ITEM_CLICK_REPORT_TOPIC, Produced.with(Serdes.String(), Serdes.Long())));
        try {
            kafkaStreams = new KafkaStreams(streamsBuilder.build(), KafkaConfigFactory.getStreamProperties());
            kafkaStreams.start();
        } catch (Exception e) {
            kafkaStreams.close();
            e.printStackTrace();
        }
    }

    public void startConsumeClickReport() {
        kafkaConsumer.subscribe(List.of(ITEM_CLICK_REPORT_TOPIC));
        TimerTask task = timeout -> {
            while (true) {
                final ConsumerRecords<String, Long> records = kafkaConsumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, Long> record : records) {
                    System.out.println(String.format("consume item_click_report_topic message, second window = %d, key = %s, value = %d", Long.parseLong(record.key().split("#")[0])/1000, record.key().split("#")[1], record.value()));
                }
            }
        };
        hashedWheelTimer.newTimeout(task, 1000, TimeUnit.MILLISECONDS);
    }

    public void stopConsumeClickReport() {
        TimerTask task = timeout ->  kafkaConsumer.close();
        hashedWheelTimer.newTimeout(task, 100, TimeUnit.MILLISECONDS);
    }

    public void stopClickStreamProcess() {
        if (NOT_RUNNING != kafkaStreams.state()) {
            kafkaStreams.close();
            System.out.println("stop kafka stream success...");
        }
    }
}
